package com.seu.me.zookeeper.curator.base;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.BackgroundCallback;
import org.apache.curator.framework.api.CuratorEvent;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Curator是Netflix公司开源的一套Zookeeper客户端框架，和ZkClient一样，解决了非常底层的细节开发工作，
 * 包括连接重连、反复注册Watcher和NodeExistsException异常等。
 * 目前已经成为Apache的顶级项目。
 * 另外还提供了一套易用性和可读性更强的Fluent风格的客户端API框架。
 * 除此之外，Curator中还提供了Zookeeper各种应用场景（Recipe，如共享锁服务、Master选举机制和分布式计算器等）的抽象封装。
 *
 * @author liangfeihu
 * @since 2017/12/5 19:27.
 */
public class CuratorBase {

    /**
     * zookeeper地址
     */
    static final String CONNECT_ADDR = "192.168.169.129:2181,192.168.169.130:2181,192.168.169.131:2181";
    /**
     * session超时时间 ms
     */
    static final int SESSION_TIMEOUT = 50000;

    /**
     * mian 测试
     *
     * @param args
     * @throws Exception
     */
    public static void main(String[] args) throws Exception {

        //1、重试策略：初试时间为1s 重试10次
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 10);
        //2、通过工厂创建连接
        CuratorFramework cf = CuratorFrameworkFactory.builder()
                .connectString(CONNECT_ADDR)
                .sessionTimeoutMs(SESSION_TIMEOUT)
                .connectionTimeoutMs(30000)
                .retryPolicy(retryPolicy)
//					.namespace("super")
                .build();
        //3、开启连接
        cf.start();

        System.out.println("ZooKeeper.States.CONNECTED=" + ZooKeeper.States.CONNECTED);
        System.out.println("State=" + cf.getState());

        //4 建立节点 指定节点类型（不加withMode默认为持久类型节点）、路径、数据内容
        cf.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath("/super1/c1", "c1内容".getBytes());
        //5 删除节点
        cf.delete().guaranteed().deletingChildrenIfNeeded().forPath("/super1");

        // 创建节点
        cf.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath("/super1/c1", "c1内容".getBytes());
        cf.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath("/super1/c2", "c2内容".getBytes());
        // 读取节点
        String ret1 = new String(cf.getData().forPath("/super1/c2"));
        System.out.println(ret1);
        // 修改节点
        cf.setData().forPath("/super1/c2", "修改c2内容".getBytes());
        String ret2 = new String(cf.getData().forPath("/super1/c2"));
        System.out.println(ret2);

        // 绑定回调函数
        ExecutorService pool = Executors.newCachedThreadPool();
        cf.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT)
                .inBackground(new BackgroundCallback() {
                    @Override
                    public void processResult(CuratorFramework cf, CuratorEvent ce) throws Exception {
                        System.out.println("code:" + ce.getResultCode());
                        System.out.println("type:" + ce.getType());
                        System.out.println("线程为:" + Thread.currentThread().getName());
                    }
                }, pool)
                .forPath("/super1/c3", "c3内容".getBytes());
        //Thread.sleep(Integer.MAX_VALUE);
        Thread.sleep(5000);

        // 读取子节点getChildren方法 和 判断节点是否存在checkExists方法
        List<String> list = cf.getChildren().forPath("/super1");
        for (String p : list) {
            System.out.println("/super1/" + p);
        }

        Stat stat = cf.checkExists().forPath("/super1/c3");
        System.out.println(stat);

        Thread.sleep(2000);
        cf.delete().guaranteed().deletingChildrenIfNeeded().forPath("/super1");

    }
}
